package scalapackage.testspark

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by Germmy on 2018/5/19.
  */
object TestIpCount {

  def main(args: Array[String]) {

    val sparkConf: SparkConf = new SparkConf().setAppName("TestIpCount").setMaster("local[*]")
    val sc: SparkContext = new SparkContext(sparkConf)

    val ips: RDD[String] = sc.textFile("D:\\document\\ip\\myip.txt")

    //1、对ips加工，将关键字段提取出来为我所用，其它垃圾信息直接丢弃
    val newIps: Array[(String, String, String)] = ips.map(x => {
      val split: Array[String] = x.split("\\|")
      val ipStart = split(0)
      val ipEnd = split(1)
      val province = split(4)
      (ipStart, ipEnd, province)
    }).collect()

    //将加工过的ips打到各个集群,以后要用ip信息直接从broadcast中取即可
    val broadcast: Broadcast[Array[(String, String, String)]] = sc.broadcast(newIps)

    //2、遍历所有日志。然后用每一行的ip去集群中搜索，找出归属地
    //2.1 从文件中读出所有日志
    val httpLog: RDD[String] = sc.textFile("D:\\document\\ip\\httplog.txt")

    val newHttpLog = httpLog.map(x => {
      val split: Array[String] = x.split("\\|")//注意按照竖线分隔不能是简单的|,而应该是\\|或者[|],因为这是正则表达式
      val ip = split(1)
      //调用ip2Long
      val ipWithLong = ip2Long(ip)
      //调用binarySearch
      val idx = binarySearch(newIps, ipWithLong)
      var guishudi = "未知地区" //我估计可能就是这个玩意可能有问题
      //假如得出结果是云南，将云南打入此行中，大功告成
      if (idx != -1) {
        guishudi = broadcast.value(idx)._3
      }
      (guishudi, 1)
    })


    //    3、reduce,针对newHttpLog reduce
    val results: RDD[(String, Int)] = newHttpLog.reduceByKey(_ + _)
    println(results.collect().toBuffer)

    //    4、写入mysql-将上面的完成，我在docker里面将mysql启动。
  }

  //scala中2个参数怎么搞？
  def binarySearch(arr: Array[(String, String, String)], num: Long): Int = {
    var startIndex = 0
    var endIndex = arr.length - 1;
    //注意这个地方必须是arr.length-1,如果是arr.length的话，那么startIndex就有可能变成arr.length,此时再调用就会数组越界
    var index = 0;

    while (startIndex <= endIndex) {
      index = (startIndex + endIndex) / 2
      if (num >= arr(index)._1.toLong && num <= arr(index)._2.toLong) {
        return index//必须return,否则下面的代码还是会走
      }

      if (num > arr(index)._2.toLong) {
        startIndex = index + 1
      } else {
        endIndex = index - 1
      }
    }
    -1 //找不到就返回-1
  }


  /**
    * 将String类型转成Long
    *
    * @param ip
    * @return
    */
  def ip2Long(ip: String): Long = {
    val fragments = ip.split("[.]")
    var ipNum = 0L
    for (i <- 0 until fragments.length) {
      ipNum = fragments(i).toLong | ipNum << 8L
    }
    ipNum
  }


}
